package concurrent;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 基于数组的阻塞队列实现，其内部维护一个定长的数组，用于存储队列元素。
 * 线程阻塞的实现是通过ReentrantLock来完成的，数据的插入与取出共用同一个锁，
 * 因此ArrayBlockingQueue并不能实现生产、消费同时进行。
 * 而且在创建ArrayBlockingQueue时，我们还可以控制对象的内部锁是否采用公平锁，默认采用非公平锁。
 * @author jack.wu
 * @since 2019/12/23 10:40
 */
public class BlockingQueueTest {
    /**
     * 推送队列
     */
    static ArrayBlockingQueue<String> pushQueue = new ArrayBlockingQueue<>(1000);

    static {
        new Thread(() -> {
            while (true) {
                String msg;
                long startTime = System.currentTimeMillis();
                // 从推送队列获取数据
                try {
                    msg = pushQueue.take();
                    long endTime = System.currentTimeMillis();
                    // 模拟推送耗时
                    TimeUnit.MILLISECONDS.sleep(500);
                    System.out.println(String.format("[%s,%s,task耗时：%s，发送消息：%s]",
                            startTime, endTime, (endTime - startTime), msg));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    /**
     * 推送消息，需要将推送消息先放入队列中
     *
     * @param msg 消息
     * @throws InterruptedException
     */
    public static void pushMsg(String msg) throws InterruptedException {
        pushQueue.put(msg);
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 1; i <= 5; i++) {
            String msg = "消息"+i;
            // 模拟耗时
            TimeUnit.SECONDS.sleep(i);
            pushMsg(msg);
        }
    }

}
